package com.atguigu.gmall.realtime.app.dim;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.func.CheckTableFunction;
import com.atguigu.gmall.realtime.app.func.DimSinkFunction;
import com.atguigu.gmall.realtime.app.func.TableProcessFunction;
import com.atguigu.gmall.realtime.beans.TableProcess;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

public class DimApp {



    //1、kafka的数据读取到flink流中
    //2、把kafka流中的jsonstring 转为jsonObject,后面方便处理。
    //3、准备维度的配置信息表，建表，有数据定义哪些是维度表。
    //4、通过flinkcdc 读取为数据流
    //5、通过广播的方式把维度配置信息流分发到kafka流中
    //6、实现两个方法  1  实现状态的写入  2 读取状态处理维度数据（过滤、剪裁、标识）
    //7、把已经过滤、剪裁的维度数据根据标识写入hbase
    public static void main(String[] args) throws Exception {
        //0、环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // env.enableCheckpointing(3000);
        env.setParallelism(4);

        //1、kafka的数据读取到flink流中
        String topic="topic_db";
        String groupId="dim_app";
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        DataStreamSource<String> kafkaStream = env.addSource(kafkaConsumer) ;

        //2、把kafka流中的jsonstring 转为jsonObject,后面方便处理。
        SingleOutputStreamOperator<JSONObject> jsonObjDstream = kafkaStream.map(jsonString -> JSON.parseObject(jsonString))
                .filter(jsonObj->jsonObj.getString("database").equals("gmall0725"))
                .filter(jsonObj->(!jsonObj.getString("type").equals("bootstrap-start")) &&( !jsonObj.getString("type").equals("bootstrap-complete")));
        // jsonObjDstream.print("aaaa---");
        //3、准备维度的配置信息表，建表，有数据定义哪些是维度表。
        //    a 要有维度的表名 、 b 维度字段 c 目标表名

        //4、通过flinkcdc 读取为数据流
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("hadoop202")
                .port(3306)
                .databaseList("gmall0725_config") // set captured database
                .tableList("gmall0725_config.table_process") // set captured table
                .username("root")
                .password("123456")
                .startupOptions(StartupOptions.initial())
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .build();


        DataStreamSource<String> tableProcessStream = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "mysql source");
       // tableProcessStream.print();  //flinkcdc得到的数据默认只会进入一个并行度   ，如果代码中有设定了多个并行度 ，也只会进入其中一个，其他的并行度没有数据
       // tableProcessStream.rebalance();//如果数据量较大，担心倾斜问题可以使用rebalance把数据重新分配一下
        SingleOutputStreamOperator<String> tableProcessAfterCheckTableStream = tableProcessStream.map(new CheckTableFunction());

        //5、通过广播的方式把维度配置信息流分发到kafka流中
        //5.1 定义状态描述符   定义状态的结构和标识  key: source_table_name  value: table_process 对象
        MapStateDescriptor<String, TableProcess>  tableProcessMapStateDescriptor
                =new MapStateDescriptor<String, TableProcess>("table_process_state",String.class,TableProcess.class);
        //5.2 制作广播流
        BroadcastStream<String> tableProcessBroadCastStream = tableProcessAfterCheckTableStream.broadcast(tableProcessMapStateDescriptor);
        //5.3  连接广播流到主流中
        BroadcastConnectedStream<JSONObject, String> broadcastConnectedStream = jsonObjDstream.connect(tableProcessBroadCastStream);

        //6、实现两个方法  1  实现状态的写入  2 读取状态处理维度数据（过滤、剪裁、标识）
        // 三个泛型： 1  主流结构  2 广播流结构  3 处理后的最终结构
        SingleOutputStreamOperator<JSONObject> afterProcessStream = broadcastConnectedStream.process(  new TableProcessFunction(tableProcessMapStateDescriptor)  );


        afterProcessStream.print("处理后的数据>>>>");

        //7、把已经过滤、剪裁的维度数据根据标识写入hbase

        afterProcessStream.addSink(new DimSinkFunction());
        env.execute();
    }
}
